package com.zc.finkDemo;

import com.alibaba.fastjson2.JSON;
import com.zc.finkDemo.entities.ErrorMessage;
import com.zc.finkDemo.entities.RelationInfo;
import com.zc.finkDemo.entities.RuleInfo;
import com.zc.finkDemo.service.RelationService;
import com.zc.finkDemo.service.RuleService;
import com.zc.finkDemo.sink.KafkaSink;
import com.zc.finkDemo.source.KafkaSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Push {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //模拟暂时设置为1
        env.setParallelism(1);
        //定义数据源
        DataStream<ErrorMessage> kafkaDS = env.addSource(new KafkaSource());

        SingleOutputStreamOperator<String> map = kafkaDS.map(errorMessage -> {
            System.out.println("收到消息：" + JSON.toJSONString(errorMessage));
            //查询规则
            RuleInfo rule = RuleService.getRule("传入查询规则的key 或者条件");
            //根据规则条件 判断是否需要推送  如果满足 根据宿主机和内部服务的关系  拼接推送内容并推送
            //查询关系
            RelationInfo relation = RelationService.getRelation("传入查询关系的key 或者条件");
            String pushContent = "触发规则:" + rule.getName() + "  影响实例：" + errorMessage.getParentECS() + " 关联影响：" + relation;
            return "告警内容";
        });
        //把结果输出
        map.addSink(new KafkaSink());

        env.execute("processErrorJob");
    }
}
